package org.lxy.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.lxy.config.RocketMQConfig;
import org.lxy.core.exception.BusException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.stream.Collectors;

/**
 * @author 卢心怡
 */
@ConditionalOnBean(RocketMQConfig.class)
@Component
@Slf4j
public class RocketUtils {

    @Autowired
    private DefaultMQProducer defaultMQProducer;

    public SendResult send(String topic, String tags, String body) {
        try {
            SendResult send = defaultMQProducer.send(Arrays.asList(tags.split(",")).stream().map(e -> new Message(topic, e, body.getBytes())).collect(Collectors.toList()));
            log.info("rocketmq send success,param[topic:{},tags:{},body:{}] result:[{}]", topic, tags, body, send);
            return send;
        } catch (Exception e) {
            log.error("rocketmq send error, param[topic:{},tags:{},body:{}]", topic, tags, body, e);
        }
        throw new BusException("rocketmq send error");
    }





    public void sendCallBack(String topic, String tags, String body, SendCallback sendCallback) {
        try {
            defaultMQProducer.send(new Message(topic, tags, body.getBytes()),sendCallback);
            log.info("rocketmq sendCallBack success,param[topic:{},tags:{},body:{}]", topic, tags, body);
        } catch (Exception e) {
            log.error("rocketmq send error, param[topic:{},tags:{},body:{}]", topic, tags, body, e);
            throw new BusException("rocketmq sendCallBack error");
        }
    }


    /**
     * @param topic
     * @param tags
     * @param body
     * @param delayLevel "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h" 对应rocket每个延迟等级的时间
     * @return
     */
    public SendResult sendDelay(String topic, String tags, String body, int delayLevel) {
        try {
            Message message = new Message(topic, tags, body.getBytes());
            message.setDelayTimeLevel(delayLevel);
            SendResult send = defaultMQProducer.send(message);
            log.info("rocketmq sendDelay success,param[topic:{},tags:{},body:{},delayLevel:{}] result:[{}]", topic, tags, body,delayLevel, send);
            return send;
        } catch (Exception e) {
            log.error("rocketmq sendDelay error, param[topic:{},tags:{},body:{}]", topic, tags, body, e);
        }
        throw new BusException("rocketmq sendDelay error");
    }

}
